package com.atguigu.chapter11;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.junit.Before;
import org.junit.Test;

/**
 * @ClassName: Flink11_Group_Window_Use
 * @Description:
 * @Author: kele
 * @Date: 2021/4/13 12:39
 *
 * 分组窗口的应用
 *
 *    滚动窗口
 *      .window( Tumble.over(lit(5).second()).on($("ts")).as("win") )
 *
 *    滑动窗口
 *      window( Slide.over( lit(5).second() ).every( lit(2).second() ).on($("ts")) .as("swin") )
 *
 *    会话窗口
 *      .window(Session.withGap(lit(2).second()).on($("ts")).as("sewin") )
 **/
public class Flink11_Group_Window_SQL_Use {

    private StreamTableEnvironment tenv;
    private StreamExecutionEnvironment env;


    @Before
    public void before(){

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(1);

        tenv = StreamTableEnvironment.create(env);

    }

    /**
     *  滚动窗口查询数据
     */

    @Test
    public void TumbleWindow() {

        //从文本文件中读取数据
        tenv.executeSql("create table senion(\n" +
                "   id string,\n" +
                "   ts bigint,\n" +
                "   vc int,\n" +
                //读取数据，用事件时间作为水印
                "   t as to_timestamp( from_unixtime( ts,'yyyy-MM-dd HH:mm:ss') ),\n" +
                //设置水印
                "   watermark for t as t -interval '5' second\n" +
                "   )with(\n" +
                "   'connector' = 'filesystem',\n" +
                "   'path' = 'in/sensor.txt', \n" +
                "   'format' = 'csv'\n" +
                "   )");

        //根据窗口查询数据
        tenv.sqlQuery("select\n" +
                "id,\n" +
                //打印窗口的起始时间
                "tumble_start(t, interval '5' second),\n" +
                //打印窗口的结束时间
                "tumble_end(t, interval '5' second),\n" +
                "sum(vc) vc_sum \n" +
                "from senion\n" +
                //按照窗口和id聚合
                "group by id,tumble(t,interval '5' second)")
                .execute()
                .print();

    }


    /**
     * 滑动窗口创建数据
     */
    @Test
    public void SlideSql() throws Exception {

        /**
         *   从kafka读取数据
         *   水印设置5s
         */

        tenv.executeSql("create table IK(\n" +
                "id string,\n" +
                "ts bigint,\n" +
                "vc int,\n" +
                "t as to_timestamp(from_unixtime(ts,'yyyy-MM-dd HH:mm:ss')),\n" +
                "watermark for t as t -interval '5' second\n" +
                ")with(\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'senior',\n" +
                "  'properties.bootstrap.servers' = 'hadoop162:9092',\n" +
                "  'properties.group.id' = 'IK',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'csv'\n" +
                ")");

//        tenv.sqlQuery("select * from IK").execute().print();

        //从文本文件中读取数据
        /*tenv.executeSql("create table IK(\n" +
                "   id string,\n" +
                "   ts bigint,\n" +
                "   vc int,\n" +
                //读取数据，用事件时间作为水印
                "   t as to_timestamp( from_unixtime( ts,'yyyy-MM-dd HH:mm:ss') ),\n" +
                //设置水印
                "   watermark for t as t -interval '5' second\n" +
                "   )with(\n" +
                "   'connector' = 'filesystem',\n" +
                "   'path' = 'in/sensor.txt', \n" +
                "   'format' = 'csv'\n" +
                "   )");*/

        /**
         * 滑动窗口，  步长5s，窗口长度10s
         */
       tenv.sqlQuery("select \n" +
                "   id, \n" +
                "   HOP_START(t, interval '5' second, interval '10' second) starttime, \n" +
                "   HOP_END(t, interval '5' second, interval '10' second) starttime, \n" +
                "   sum(vc) vc_sum \n" +
                "   from IK \n" +
                "   group by id,HOP(t, interval '5' second, interval '10' second)")
                .execute()
                .print();

        env.execute();

    }


    /**
     * 会话窗口
     */

    @Test
    public void SessionWindow() {

        //从文本文件中读取数据
        tenv.executeSql("create table senion(\n" +
                "   id string,\n" +
                "   ts bigint,\n" +
                "   vc int,\n" +
                //读取数据，用事件时间作为水印
                "   t as to_timestamp( from_unixtime( ts,'yyyy-MM-dd HH:mm:ss') ),\n" +
                //设置水印
                "   watermark for t as t -interval '5' second\n" +
                "   )with(\n" +
                "   'connector' = 'filesystem',\n" +
                "   'path' = 'in/sensor.txt', \n" +
                "   'format' = 'csv'\n" +
                "   )");

        //根据窗口查询数据
        tenv.sqlQuery("select\n" +
                "id,\n" +
                //打印窗口的起始时间
                "SESSION_START(t, interval '2' second),\n" +
                //打印窗口的结束时间
                "SESSION_START(t, interval '2' second),\n" +
                "sum(vc) vc_sum \n" +
                "from senion\n" +
                //按照窗口和id聚合
                "group by id,SESSION(t, interval '2' second)")
                .execute()
                .print();

    }




}
